package storm;

import org.apache.storm.task.OutputCollector;
import org.apache.storm.task.TopologyContext;
import org.apache.storm.topology.OutputFieldsDeclarer;
import org.apache.storm.topology.base.BaseRichBolt;
import org.apache.storm.tuple.Fields;
import org.apache.storm.tuple.Tuple;
import org.apache.storm.tuple.Values;

import java.util.Map;

/**
 * @author Hanhan_Jun
 * @date 2021年12月11日 13:46
 */
public class Bolt2 extends BaseRichBolt {
    //    private static final long serialVersionUID = 886149197481637894L;
    private OutputCollector collector;

    public void prepare(Map stormConf, TopologyContext context, OutputCollector collector) {
        this.collector = collector;
    }

    public void execute(Tuple input) {
        String orderdetail_id = input.getStringByField("orderdetail_id");
        String orderid = input.getStringByField("orderid");
        String sender = input.getStringByField("sender");
        String sendplace = input.getStringByField("sendplace");
        String recipient = input.getStringByField("recipient");
        String receiveplace = input.getStringByField("receiveplace");
        String methodofpayment = input.getStringByField("methodofpayment");
        String paytime = input.getStringByField("paytime");
        Double freight = input.getDoubleByField("freight");
        Double total = input.getDoubleByField("total");

        collector.emit(new Values(recipient, receiveplace, total,methodofpayment,paytime));
//        System.out.println("orderdetail_id:"+orderdetail_id+" orderid:"+orderid+" sender:"+sender+" sendplace = " + sendplace
//                +" recipient:"+recipient+" receiveplace :"+receiveplace +" methodofpayment:"+methodofpayment+" paytime:"+paytime+" freight:"+ freight+" total:"+total);
    }

    public void declareOutputFields(OutputFieldsDeclarer declarer) {
        declarer.declare(new Fields("recipient", "receiveplace", "total","methodofpayment","paytime"));
    }
}

